package com.winit.schedule.task.tracker;

import java.io.IOException;
import java.util.Date;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import com.alibaba.fastjson.JSONObject;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.task.TaskInterfaceType;
import org.quartz.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.event.AsyncDispatcher;
import com.winit.schedule.core.event.Dispatcher;
import com.winit.schedule.core.model.AssignedTask;
import com.winit.schedule.core.model.TaskID;
import com.winit.schedule.core.repository.constants.QuartzManagerType;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.service.CompositeService;
import com.winit.schedule.core.task.ScheduleTaskDefine;
import com.winit.schedule.core.task.ScheduleTaskScanner;
import com.winit.schedule.core.utils.Md5Encrypt;
import com.winit.schedule.core.utils.NetUtils;
import com.winit.schedule.task.tracker.conf.TaskConfiguration;
import com.winit.schedule.task.tracker.event.ConfigChangeEventType;
import com.winit.schedule.task.tracker.monitor.ExpireTaskRunnerMonitor;
import com.winit.schedule.task.tracker.monitor.HeartbeatMonitor;
import com.winit.schedule.task.tracker.monitor.ZKServiceMonitor;
import com.winit.schedule.task.tracker.quartz.QuartzScheduleService;
import com.winit.schedule.task.tracker.remoting.RemotingClientService;
import com.winit.schedule.task.tracker.runner.DefaultTaskManager;
import com.winit.schedule.task.tracker.runner.StrategyTask;
import com.winit.schedule.task.tracker.service.RPCRetryService;

public class TaskTracker extends CompositeService {

    private static Logger                   logger      = LoggerFactory.getLogger(TaskTracker.class);

    private Map<TaskID, List<StrategyTask>> managerMap  = new ConcurrentHashMap<TaskID, List<StrategyTask>>();

    private Set<ScheduleTaskDefine>         taskDefines = new HashSet<ScheduleTaskDefine>();

    private String                          uuid;

    private String                          ip;

    private String                          hostName;

    private Configuration                   conf;

    private Dispatcher                      dispatcher;

    private TaskTrackerContextImpl          taskTrackerContext;

    private HeartbeatMonitor                heartbeatMonitor;

    private ApplicationContext              applicationContext;

    private RemotingClientService           remotingClientService;

    private ExpireTaskRunnerMonitor         expireTaskRunnerMonitor;

    private String[]                        scanPackages;

    private TaskTrackerEntity               taskTrackerEntity;

    private List<Scheduler>                 quartzSchedulers;

    private RPCRetryService                 rpcRetryService;

    public TaskTracker(){
        this("TaskTracker");
    }

    public TaskTracker(String name){
        super(name);
        this.ip = NetUtils.getLocalIP();
        this.hostName = NetUtils.getLocalHostName();
    }

    @Override
    protected void serviceInit(Configuration conf) throws Exception {
        this.conf = conf;

        boolean enabled = conf.getBoolean(TaskConfiguration.USCHEDULE_ENABLE,
            TaskConfiguration.DEFAULT_USCHEDULE_ENABLE);

        if (enabled) {
            String taskGroup = conf.getString(TaskConfiguration.JT_TASK_GROUP, TaskConfiguration.DEFAULT_JT_TASK_GROUP);
            String uuidRandom = conf.getString(TaskConfiguration.UUID_RANDOM, TaskConfiguration.DEFAULT_UUID_RANDOM);
            String source = this.getClass().getResource("/").getPath() + uuidRandom;
            this.uuid = this.ip + "$" + taskGroup + "$" + Md5Encrypt.md5(source);

            this.taskTrackerContext = new TaskTrackerContextImpl();
            taskTrackerContext.setTaskTracker(this);

            // 扫描获取任务定义bean
            this.taskDefines = scanTaskDefineBean();
            taskTrackerContext.setTaskDefines(taskDefines);

            // 初始化taskTrackerEntity
            initTaskTrackerInfo(conf);

            // 构建事件分发器
            this.dispatcher = setupDispatcher();
            addIfService(dispatcher);
            taskTrackerContext.setDispatcher(dispatcher);

            // zk
            ZKServiceMonitor zkServiceMonitor = new ZKServiceMonitor(this.taskTrackerContext);
            addIfService(zkServiceMonitor);
            taskTrackerContext.setZkServiceMonitor(zkServiceMonitor);

            // RPC请求重试
            rpcRetryService = new RPCRetryService(this.taskTrackerContext);
            addIfService(rpcRetryService);
            this.taskTrackerContext.setRpcRetryService(rpcRetryService);

            // 构建远程服务
            remotingClientService = new RemotingClientService(taskTrackerContext);
            addIfService(remotingClientService);
            taskTrackerContext.setRemotingClientService(remotingClientService);
            dispatcher.register(ConfigChangeEventType.class, remotingClientService);

            // 构建心跳监控服务
            this.heartbeatMonitor = new HeartbeatMonitor(taskTrackerContext, remotingClientService);
            addIfService(heartbeatMonitor);
            taskTrackerContext.setHeartbeatMonitor(heartbeatMonitor);

            this.expireTaskRunnerMonitor = new ExpireTaskRunnerMonitor(this);
            addIfService(expireTaskRunnerMonitor);
            taskTrackerContext.setExpireTaskRunnerMonitor(expireTaskRunnerMonitor);

            // Quartz Support

            if (null != quartzSchedulers && quartzSchedulers.size() > 0) {
                QuartzScheduleService quartzScheduleService = new QuartzScheduleService(quartzSchedulers,
                    taskTrackerContext);
                addIfService(quartzScheduleService);
                taskTrackerContext.setQuartzScheduleService(quartzScheduleService);
                dispatcher.register(QuartzManagerType.class, quartzScheduleService);
            }
        }

        super.serviceInit(conf);
    }

    @Override
    protected void serviceStart() throws Exception {
        super.serviceStart();
    }

    @Override
    protected void serviceStop() throws Exception {
        super.serviceStop();
        forceStopAllTask();
    }

    private void initTaskTrackerInfo(Configuration conf) {
        String taskGroup = conf.getString(TaskConfiguration.JT_TASK_GROUP, TaskConfiguration.DEFAULT_JT_TASK_GROUP);

        logger.info("The parameters of task group is : " + taskGroup);

        if (TaskConfiguration.DEFAULT_JT_TASK_GROUP.equals(taskGroup)) {
            logger.warn("The parameters of task group is the default value, there may be hidden dangers of task conflict");
        }

        taskTrackerEntity = new TaskTrackerEntity();
        taskTrackerEntity.setHostname(this.hostName);
        taskTrackerEntity.setIp(this.ip);
        taskTrackerEntity.setTaskGroup(taskGroup);
        taskTrackerEntity.setTaskTrackerUuid(this.uuid);
        taskTrackerEntity.setTaskDefines(this.taskDefines);
    }

    /**
     * Register the handlers for alwaysOn services
     */
    private Dispatcher setupDispatcher() {
        Dispatcher dispatcher = new AsyncDispatcher();
        return dispatcher;
    }

    /**
     * 扫描注解
     * 
     * @return
     * @throws IOException
     * @throws ClassNotFoundException
     */
    private Set<ScheduleTaskDefine> scanTaskDefineBean() throws IOException, ClassNotFoundException {
        String taskGroup = conf.getString(TaskConfiguration.JT_TASK_GROUP, TaskConfiguration.DEFAULT_JT_TASK_GROUP);
        List<String> excludes = (List<String>) conf.getStringCollection(TaskConfiguration.JT_TASK_EXCLUDE);
        ScheduleTaskScanner scanner = new ScheduleTaskScanner(taskGroup, scanPackages, excludes);
        return scanner.scan();
    }

    /**
     * 停止任务
     * 
     * @param taskIds
     */
    public void stopTask(List<TaskID> taskIds) {
        if (null != taskIds) {
            for (TaskID taskId : taskIds) {
                List<StrategyTask> taskManagers = managerMap.get(taskId);
                if (null != taskManagers) {
                    for (StrategyTask taskManager : taskManagers) {
                        taskManager.stop(false);
                    }
                }
            }
        }
    }

    /**
     * 停止所有任务
     */
    public void forceStopAllTask() {
        if (logger.isDebugEnabled()) {
            logger.debug("--------------------Stop all task--------------------------");
        }
        TaskID[] taskIdList = (TaskID[]) this.managerMap.keySet().toArray(new TaskID[0]);
        for (TaskID taskId : taskIdList) {
            List<StrategyTask> taskManagers = managerMap.get(taskId);
            if (null != taskManagers) {
                for (StrategyTask taskManager : taskManagers) {
                    try {
                        taskManager.stop(true);
                    } catch (Exception e) {
                        logger.error("Stop task failure:" + e.getMessage(), e);
                    }
                }
            }
        }
    }

    /**
     * 停止所有任务
     */
    public void stopAllTask() {
        if (logger.isDebugEnabled()) {
            logger.debug("--------------------Stop all task--------------------------");
        }
        TaskID[] taskIdList = (TaskID[]) this.managerMap.keySet().toArray(new TaskID[0]);
        for (TaskID taskId : taskIdList) {
            List<StrategyTask> taskManagers = managerMap.get(taskId);
            if (null != taskManagers) {
                for (StrategyTask taskManager : taskManagers) {
                    try {
                        taskManager.stop(false);
                    } catch (Exception e) {
                        logger.error("Stop task failure:" + e.getMessage(), e);
                    }
                }
            }
        }
    }

    /**
     * 启动任务
     * 
     * @param tasks
     */
    public synchronized void startTask(List<AssignedTask> tasks) {
        if (null != tasks) {
            for (AssignedTask task : tasks) {
                try {
                    List<StrategyTask> taskManagers = managerMap.get(task.getTaskId());

                    if (null == taskManagers) {
                        taskManagers = new ArrayList<StrategyTask>();
                        managerMap.put(task.getTaskId(), taskManagers);
                    }

                    taskManagers.add(new DefaultTaskManager(this, task));

                    if (logger.isDebugEnabled()) {
                        logger.debug("Startup taskRunner success. task id :" + task.getTaskId().getId()
                                     + ", task item : " + task.getTaskItems());
                    }

                } catch (Exception e) {
                    logger.error("Startup taskRunner failure. task id :" + task.getTaskId().getId() + ", task item : "
                                 + task.getTaskItems());
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * 启动任务
     * 
     * @param tasks
     */
    public synchronized void reAllocationTask(List<AssignedTask> tasks) {
        if (null != tasks) {
            for (AssignedTask task : tasks) {
                try {
                    List<StrategyTask> taskManagers = managerMap.get(task.getTaskId());

                    if (null == taskManagers) {
                        taskManagers = new ArrayList<StrategyTask>();
                        managerMap.put(task.getTaskId(), taskManagers);
                    }

                    if (task.getCurrentFetchNum() > 0) {
                        // 抓取的任务项数目大于0，表示需要新执行的任务
                        taskManagers.add(new DefaultTaskManager(this, task));
                    } else {
                        // 抓取的任务项数目小于0，表示当前JVM运行任务项数量过载，需要停止运行的TaskRunner数量
                        for (int i = 1; i <= Math.abs(task.getCurrentFetchNum()); i++) {
                            int size = taskManagers.size() - i;
                            if (size >= 0 && size < taskManagers.size()) {
                                taskManagers.get(size).stop(false);
                            }
                        }
                    }

                    if (logger.isDebugEnabled()) {
                        logger.debug("Startup taskRunner success. task id :" + task.getTaskId().getId()
                                     + ", task item : " + task.getTaskItems());
                    }

                } catch (Exception e) {
                    logger.error("Startup taskRunner failure. task id :" + task.getTaskId().getId() + ", task item : "
                                 + task.getTaskItems());
                    e.printStackTrace();
                }
            }
        }

    }

    public synchronized void removeTaskManager(TaskID taskId, StrategyTask strategyTask) {
        List<StrategyTask> taskManagers = managerMap.get(taskId);
        taskManagers.remove(strategyTask);
        if (taskManagers.isEmpty()) {
            managerMap.remove(taskId);
        }
    }

    public Configuration getConf() {
        return conf;
    }

    public void setConf(Configuration conf) {
        this.conf = conf;
    }

    public Dispatcher getDispatcher() {
        return dispatcher;
    }

    public void setDispatcher(Dispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }

    public TaskTrackerContextImpl getTaskTrackerContext() {
        return taskTrackerContext;
    }

    public void setTaskTrackerContext(TaskTrackerContextImpl taskTrackerContext) {
        this.taskTrackerContext = taskTrackerContext;
    }

    public Map<TaskID, List<StrategyTask>> getManagerMap() {
        return managerMap;
    }

    public void setManagerMap(Map<TaskID, List<StrategyTask>> managerMap) {
        this.managerMap = managerMap;
    }

    public String getUuid() {
        return uuid;
    }

    public String getIp() {
        return ip;
    }

    public String getHostName() {
        return hostName;
    }

    public HeartbeatMonitor getHeartbeatMonitor() {
        return heartbeatMonitor;
    }

    public ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    public Object getBean(String beanName) {
        if (null != applicationContext) {
            return applicationContext.getBean(beanName);
        }
        return null;
    }

    public String[] getScanPackages() {
        return scanPackages;
    }

    public void setScanPackages(String[] scanPackages) {
        this.scanPackages = scanPackages;
    }

    public Set<ScheduleTaskDefine> getTaskDefines() {
        return taskDefines;
    }

    public TaskTrackerEntity getTaskTrackerEntity() {
        return taskTrackerEntity;
    }

    public RemotingClientService getRemotingClientService() {
        return remotingClientService;
    }

    public List<Scheduler> getQuartzSchedulers() {
        return quartzSchedulers;
    }

    public void setQuartzSchedulers(List<Scheduler> quartzSchedulers) {
        this.quartzSchedulers = quartzSchedulers;
    }

    public RPCRetryService getRpcRetryService() {
        return rpcRetryService;
    }



    public void taskRun(String taskGroup,String taskName,String taskClass,String tId,String taskInterfaceType,String taskParameter,String tItem,String uId){
        List<AssignedTask> tasks = new ArrayList<AssignedTask>();
        AssignedTask task = new AssignedTask();
//        us_task_define
        ScheduleTaskDefine taskDefine = new ScheduleTaskDefine();
        taskDefine.setGroup(taskGroup);
        taskDefine.setTaskName(taskName);
        taskDefine.setTaskDealClass(taskClass);
        if(TaskInterfaceType.SCHEDULE_TASK_SINGLE.name().equals(taskInterfaceType)){
            taskDefine.setInterfaceType(TaskInterfaceType.SCHEDULE_TASK_SINGLE);
        }else{
            taskDefine.setInterfaceType(TaskInterfaceType.SCHEDULE_TASK_MUTIL);
            task.setTaskParameter(taskParameter);
            task.setTaskItem(tItem);
        }
        task.setTaskDefine(taskDefine);
//      us_task_item
        List<TaskItemEntity> taskItems = new ArrayList<TaskItemEntity>();
        TaskItemEntity taskItem = new TaskItemEntity();
        TaskID taskId = new TaskID();
        taskId.setTaskId(tId);
        taskItem.setTaskId(taskId);
        taskItems.add(taskItem);
        task.setTaskItems(taskItems);
        task.setTaskId(taskId);
        task.setTaskGroup("MatrixSend_ON");
        task.setTaskName(uId);


        tasks.add(task);

        TaskTracker taskTracker = new TaskTracker();
        taskTracker.reAllocationTask(tasks);
    }
}
